{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "d99f1cb7",
   "metadata": {},
   "source": [
    "# Using Tester to test FastKafka"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "675bd571",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | hide\n",
    "\n",
    "from fastkafka._application.app import FastKafka\n",
    "from fastkafka._application.tester import Tester\n",
    "from pydantic import BaseModel, Field\n",
    "from typing import List, Optional\n",
    "from IPython.display import Markdown, display_markdown"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "77293df7",
   "metadata": {},
   "source": [
    "In order to speed up development and make testing easier, we have implemented the [Tester](../../api/fastkafka/testing/Tester/) class. \n",
    "\n",
    "The [Tester](../../api/fastkafka/testing/Tester/) instance starts in-memory implementation of Kafka broker i.e. there is no need for starting localhost Kafka service for testing FastKafka apps.\n",
    "The [Tester](../../api/fastkafka/testing/Tester/) will redirect `consumes` and `produces` decorated functions to the in-memory Kafka broker so that you can quickly test FasKafka apps without the need of a running Kafka broker and all its dependencies. Also, for each FastKafka `consumes` and `produces` function, [Tester](../../api/fastkafka/testing/Tester/) will create it's mirrored fuction i.e. if the `consumes` function is implemented, the [Tester](../../api/fastkafka/testing/Tester/) will create the `produces` function (and the other way - if the `produces` function is implemented, [Tester](../../api/fastkafka/testing/Tester/) will create `consumes` function)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3e9ea6d4",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | hide\n",
    "# | notest\n",
    "\n",
    "# allows async calls in notebooks\n",
    "\n",
    "import nest_asyncio"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9cde8241",
   "metadata": {},
   "outputs": [],
   "source": [
    "# | hide\n",
    "# | notest\n",
    "\n",
    "nest_asyncio.apply()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1278854a",
   "metadata": {},
   "source": [
    "## Basic example\n",
    "\n",
    "To showcase the functionalities of FastKafka and illustrate the concepts discussed, we can use a simple test message called `TestMsg`. Here's the definition of the `TestMsg` class:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c7754a58",
   "metadata": {},
   "outputs": [],
   "source": [
    "class TestMsg(BaseModel):\n",
    "    msg: str = Field(...)\n",
    "\n",
    "\n",
    "test_msg = TestMsg(msg=\"signal\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d15934a5",
   "metadata": {},
   "source": [
    "In this example we have implemented `FastKafka` app with one `consumes` and one `produces` function. `on_input` function consumes messages from the `input` topic and `to_output` function produces messages to the `output` topic.\n",
    "\n",
    "**Note**: it is necessary to define parameter and return types in the produces and consumes functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "27b9417e",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pydantic import BaseModel, Field\n",
    "\n",
    "app = FastKafka()\n",
    "\n",
    "\n",
    "@app.consumes()\n",
    "async def on_input(msg: TestMsg):\n",
    "    await to_output(TestMsg(msg=f\"Hello {msg.msg}\"))\n",
    "\n",
    "\n",
    "@app.produces()\n",
    "async def to_output(msg: TestMsg) -> TestMsg:\n",
    "    return msg"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4b1ce364",
   "metadata": {},
   "source": [
    "### Testing the application"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c0155687",
   "metadata": {},
   "source": [
    "In this example `app` has imlemented `on_input` and `to_output` functions. We can now use [Tester](../../api/fastkafka/testing/Tester/) to create their mirrored functions: `to_input` and `on_output`. \n",
    "\n",
    "Testing process for this example could look like this:\n",
    "\n",
    "1. `tester` produces the message to the `input` topic\n",
    "\n",
    "2. Assert that the `app` consumed the message by calling `on_input` with the accurate argument\n",
    "\n",
    "3. Within `on_input` function, `to_output` function is called - and message is produced to the `output` topic\n",
    "\n",
    "4. Assert that the `tester` consumed the message by calling `on_output` with the accurate argument"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0371097d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23-07-31 10:38:30.810 [INFO] fastkafka._testing.in_memory_broker: InMemoryBroker._patch_consumers_and_producers(): Patching consumers and producers!\n",
      "23-07-31 10:38:30.811 [INFO] fastkafka._testing.in_memory_broker: InMemoryBroker starting\n",
      "23-07-31 10:38:30.812 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': 'localhost:9092'}'\n",
      "23-07-31 10:38:30.812 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched start() called()\n",
      "23-07-31 10:38:30.826 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': 'localhost:9092'}'\n",
      "23-07-31 10:38:30.827 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched start() called()\n",
      "23-07-31 10:38:30.827 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "23-07-31 10:38:30.828 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'auto_offset_reset': 'earliest', 'max_poll_records': 100, 'bootstrap_servers': 'localhost:9092'}\n",
      "23-07-31 10:38:30.828 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched start() called()\n",
      "23-07-31 10:38:30.829 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "23-07-31 10:38:30.829 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched subscribe() called\n",
      "23-07-31 10:38:30.830 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer.subscribe(), subscribing to: ['input']\n",
      "23-07-31 10:38:30.830 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "23-07-31 10:38:30.835 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "23-07-31 10:38:30.835 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'auto_offset_reset': 'earliest', 'max_poll_records': 100, 'bootstrap_servers': 'localhost:9092'}\n",
      "23-07-31 10:38:30.836 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched start() called()\n",
      "23-07-31 10:38:30.836 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "23-07-31 10:38:30.836 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched subscribe() called\n",
      "23-07-31 10:38:30.837 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer.subscribe(), subscribing to: ['output']\n",
      "23-07-31 10:38:30.837 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "23-07-31 10:38:34.828 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched stop() called\n",
      "23-07-31 10:38:34.828 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "23-07-31 10:38:34.829 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "23-07-31 10:38:34.829 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched stop() called\n",
      "23-07-31 10:38:34.830 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched stop() called\n",
      "23-07-31 10:38:34.831 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "23-07-31 10:38:34.831 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "23-07-31 10:38:34.832 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched stop() called\n",
      "23-07-31 10:38:34.832 [INFO] fastkafka._testing.in_memory_broker: InMemoryBroker stopping\n",
      "ok\n"
     ]
    }
   ],
   "source": [
    "async with Tester(app).using_inmemory_broker() as tester:\n",
    "    input_msg = TestMsg(msg=\"Mickey\")\n",
    "\n",
    "    # tester produces message to the input topic\n",
    "    await tester.to_input(input_msg)\n",
    "    # previous line is equal to\n",
    "    # await tester.mirrors[app.on_input](input_msg)\n",
    "\n",
    "    # assert that app consumed from the input topic and it was called with the accurate argument\n",
    "    await app.awaited_mocks.on_input.assert_called_with(\n",
    "        TestMsg(msg=\"Mickey\"), timeout=5\n",
    "    )\n",
    "    # assert that tester consumed from the output topic and it was called with the accurate argument\n",
    "    await tester.on_output.assert_called_with(TestMsg(msg=\"Hello Mickey\"), timeout=5)\n",
    "print(\"ok\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1b612c56",
   "metadata": {},
   "source": [
    "### Final script"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9b94ed56",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23-07-31 10:38:34.855 [INFO] fastkafka._testing.in_memory_broker: InMemoryBroker._patch_consumers_and_producers(): Patching consumers and producers!\n",
      "23-07-31 10:38:34.856 [INFO] fastkafka._testing.in_memory_broker: InMemoryBroker starting\n",
      "23-07-31 10:38:34.856 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': 'localhost:9092'}'\n",
      "23-07-31 10:38:34.857 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched start() called()\n",
      "23-07-31 10:38:34.871 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': 'localhost:9092'}'\n",
      "23-07-31 10:38:34.872 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched start() called()\n",
      "23-07-31 10:38:34.872 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "23-07-31 10:38:34.873 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'auto_offset_reset': 'earliest', 'max_poll_records': 100, 'bootstrap_servers': 'localhost:9092'}\n",
      "23-07-31 10:38:34.874 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched start() called()\n",
      "23-07-31 10:38:34.875 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "23-07-31 10:38:34.877 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched subscribe() called\n",
      "23-07-31 10:38:34.877 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer.subscribe(), subscribing to: ['input']\n",
      "23-07-31 10:38:34.878 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "23-07-31 10:38:34.878 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "23-07-31 10:38:34.879 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'auto_offset_reset': 'earliest', 'max_poll_records': 100, 'bootstrap_servers': 'localhost:9092'}\n",
      "23-07-31 10:38:34.879 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched start() called()\n",
      "23-07-31 10:38:34.879 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "23-07-31 10:38:34.880 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched subscribe() called\n",
      "23-07-31 10:38:34.880 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer.subscribe(), subscribing to: ['output']\n",
      "23-07-31 10:38:34.881 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "23-07-31 10:38:38.873 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched stop() called\n",
      "23-07-31 10:38:38.873 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "23-07-31 10:38:38.874 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "23-07-31 10:38:38.874 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched stop() called\n",
      "23-07-31 10:38:38.875 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched stop() called\n",
      "23-07-31 10:38:38.876 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "23-07-31 10:38:38.877 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "23-07-31 10:38:38.877 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched stop() called\n",
      "23-07-31 10:38:38.878 [INFO] fastkafka._testing.in_memory_broker: InMemoryBroker stopping\n",
      "ok\n"
     ]
    }
   ],
   "source": [
    "# | notest\n",
    "\n",
    "import asyncio\n",
    "from fastkafka._application.app import FastKafka\n",
    "from fastkafka._application.tester import Tester\n",
    "from pydantic import BaseModel, Field\n",
    "\n",
    "\n",
    "class TestMsg(BaseModel):\n",
    "    msg: str = Field(...)\n",
    "\n",
    "\n",
    "app = FastKafka()\n",
    "\n",
    "\n",
    "@app.consumes()\n",
    "async def on_input(msg: TestMsg):\n",
    "    await to_output(TestMsg(msg=f\"Hello {msg.msg}\"))\n",
    "\n",
    "\n",
    "@app.produces()\n",
    "async def to_output(msg: TestMsg) -> TestMsg:\n",
    "    return msg\n",
    "\n",
    "\n",
    "async def async_tests():\n",
    "    async with Tester(app).using_inmemory_broker() as tester:\n",
    "        input_msg = TestMsg(msg=\"Mickey\")\n",
    "\n",
    "        # tester produces message to the input topic\n",
    "        await tester.to_input(input_msg)\n",
    "\n",
    "        # assert that app consumed from the input topic and it was called with the accurate argument\n",
    "        await app.awaited_mocks.on_input.assert_called_with(\n",
    "            TestMsg(msg=\"Mickey\"), timeout=5\n",
    "        )\n",
    "        # assert that tester consumed from the output topic and it was called with the accurate argument\n",
    "        await tester.awaited_mocks.on_output.assert_called_with(\n",
    "            TestMsg(msg=\"Hello Mickey\"), timeout=5\n",
    "        )\n",
    "    print(\"ok\")\n",
    "\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    loop = asyncio.get_event_loop()\n",
    "    loop.run_until_complete(async_tests())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0ca15a23",
   "metadata": {},
   "source": [
    "## Using external brokers"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7b91ccd0",
   "metadata": {},
   "source": [
    "If you have already running brokers e.g. `kafka_brokers`, you can use [Tester](../../api/fastkafka/testing/Tester/) method `using_external_broker` to set  brokers which will be used in tests. \n",
    "\n",
    "The same example as previous but with external `kafka_brokers`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "92988480",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "\n",
       "```python\n",
       "# content of the \"application_test.py\" file\n",
       "\n",
       "import asyncio\n",
       "from fastkafka._application.app import FastKafka\n",
       "from fastkafka._application.tester import Tester\n",
       "from pydantic import BaseModel, Field\n",
       "\n",
       "\n",
       "class TestMsg(BaseModel):\n",
       "    msg: str = Field(...)\n",
       "\n",
       "\n",
       "kafka_brokers = {\n",
       "    \"localhost\": {\n",
       "        \"url\": \"localhost\",\n",
       "        \"description\": \"local development kafka broker\",\n",
       "        \"port\": 9092,\n",
       "    },\n",
       "    \"production\": {\n",
       "        \"url\": \"kafka.airt.ai\",\n",
       "        \"description\": \"production kafka broker\",\n",
       "        \"port\": 9092,\n",
       "        \"protocol\": \"kafka-secure\",\n",
       "        \"security\": {\"type\": \"plain\"},\n",
       "    },\n",
       "}\n",
       "\n",
       "app = FastKafka(\n",
       "    title=\"Demo Kafka app\",\n",
       "    kafka_brokers=kafka_brokers,\n",
       ")\n",
       "\n",
       "@app.consumes()\n",
       "async def on_input(msg: TestMsg):\n",
       "    await to_output(TestMsg(msg=f\"Hello {msg.msg}\"))\n",
       "\n",
       "\n",
       "@app.produces()\n",
       "async def to_output(msg: TestMsg) -> TestMsg:\n",
       "    return msg\n",
       "\n",
       "\n",
       "async def async_tests():\n",
       "    async with Tester(app).using_external_broker(bootstrap_servers_id=\"production\") as tester:\n",
       "        input_msg = TestMsg(msg=\"Mickey\")\n",
       "\n",
       "        # tester produces message to the input topic\n",
       "        await tester.to_input(input_msg)\n",
       "\n",
       "        # assert that app consumed from the input topic and it was called with the accurate argument\n",
       "        await app.awaited_mocks.on_input.assert_called_with(\n",
       "            TestMsg(msg=\"Mickey\"), timeout=5\n",
       "        )\n",
       "        # assert that tester consumed from the output topic and it was called with the accurate argument\n",
       "        await tester.awaited_mocks.on_output.assert_called_with(\n",
       "            TestMsg(msg=\"Hello Mickey\"), timeout=5\n",
       "        )\n",
       "    print(\"ok\")\n",
       "\n",
       "\n",
       "if __name__ == \"__main__\":\n",
       "    loop = asyncio.get_event_loop()\n",
       "    loop.run_until_complete(async_tests())\n",
       "\n",
       "```\n"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# | echo: false\n",
    "\n",
    "kafka_app_source = \"\"\"\n",
    "import asyncio\n",
    "from fastkafka._application.app import FastKafka\n",
    "from fastkafka._application.tester import Tester\n",
    "from pydantic import BaseModel, Field\n",
    "\n",
    "\n",
    "class TestMsg(BaseModel):\n",
    "    msg: str = Field(...)\n",
    "\n",
    "\n",
    "kafka_brokers = {\n",
    "    \"localhost\": {\n",
    "        \"url\": \"localhost\",\n",
    "        \"description\": \"local development kafka broker\",\n",
    "        \"port\": 9092,\n",
    "    },\n",
    "    \"production\": {\n",
    "        \"url\": \"kafka.airt.ai\",\n",
    "        \"description\": \"production kafka broker\",\n",
    "        \"port\": 9092,\n",
    "        \"protocol\": \"kafka-secure\",\n",
    "        \"security\": {\"type\": \"plain\"},\n",
    "    },\n",
    "}\n",
    "\n",
    "app = FastKafka(\n",
    "    title=\"Demo Kafka app\",\n",
    "    kafka_brokers=kafka_brokers,\n",
    ")\n",
    "\n",
    "@app.consumes()\n",
    "async def on_input(msg: TestMsg):\n",
    "    await to_output(TestMsg(msg=f\"Hello {msg.msg}\"))\n",
    "\n",
    "\n",
    "@app.produces()\n",
    "async def to_output(msg: TestMsg) -> TestMsg:\n",
    "    return msg\n",
    "\n",
    "\n",
    "async def async_tests():\n",
    "    async with Tester(app).using_external_broker(bootstrap_servers_id=\"production\") as tester:\n",
    "        input_msg = TestMsg(msg=\"Mickey\")\n",
    "\n",
    "        # tester produces message to the input topic\n",
    "        await tester.to_input(input_msg)\n",
    "\n",
    "        # assert that app consumed from the input topic and it was called with the accurate argument\n",
    "        await app.awaited_mocks.on_input.assert_called_with(\n",
    "            TestMsg(msg=\"Mickey\"), timeout=5\n",
    "        )\n",
    "        # assert that tester consumed from the output topic and it was called with the accurate argument\n",
    "        await tester.awaited_mocks.on_output.assert_called_with(\n",
    "            TestMsg(msg=\"Hello Mickey\"), timeout=5\n",
    "        )\n",
    "    print(\"ok\")\n",
    "\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    loop = asyncio.get_event_loop()\n",
    "    loop.run_until_complete(async_tests())\n",
    "\"\"\"\n",
    "\n",
    "with open(\"application_test.py\", \"w\") as source:\n",
    "    source.write(kafka_app_source)\n",
    "\n",
    "Markdown(\n",
    "    f\"\"\"\n",
    "```python\n",
    "# content of the \"application_test.py\" file\n",
    "{kafka_app_source}\n",
    "```\n",
    "\"\"\"\n",
    ")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "97f3ea6b",
   "metadata": {},
   "source": [
    "## Example: New Employee app\n",
    "\n",
    "In this example, our `app` has one consumes and two produces functions.\n",
    "\n",
    "Every time a company hires an `Employee`, some employee data is sent to the `new_employee` topic.\n",
    "\n",
    "That's when our application comes into play! The app consumes this data by calling `on_new_employee`. Within this function, `to_employee_email` and `to_welcome_message` functions are called - and messages are produced to the `employee_email` and `welcome_message` topic."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f7f51841",
   "metadata": {},
   "outputs": [],
   "source": [
    "class Employee(BaseModel):\n",
    "    name: str\n",
    "    surname: str\n",
    "    email: Optional[str] = None\n",
    "\n",
    "\n",
    "class EmaiMessage(BaseModel):\n",
    "    sender: str = \"info@gmail.com\"\n",
    "    receiver: str\n",
    "    subject: str\n",
    "    message: str\n",
    "\n",
    "kafka_brokers = dict(localhost=[dict(url=\"server_1\", port=9092)], production=[dict(url=\"production_server_1\", port=9092)])\n",
    "app = FastKafka(kafka_brokers=kafka_brokers)\n",
    "\n",
    "\n",
    "@app.consumes()\n",
    "async def on_new_employee(msg: Employee):\n",
    "    employee = await to_employee_email(msg)\n",
    "    await to_welcome_message(employee)\n",
    "\n",
    "\n",
    "@app.produces()\n",
    "async def to_employee_email(employee: Employee) -> Employee:\n",
    "    # generate new email\n",
    "    employee.email = employee.name + \".\" + employee.surname + \"@gmail.com\"\n",
    "    return employee\n",
    "\n",
    "\n",
    "@app.produces()\n",
    "async def to_welcome_message(employee: Employee) -> EmaiMessage:\n",
    "    message = f\"Dear {employee.name},\\nWelcome to the company\"\n",
    "    return EmaiMessage(receiver=employee.email, subject=\"Welcome\", message=message)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7c7c6d45",
   "metadata": {},
   "source": [
    "### Testing the application"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ddcbb2a0",
   "metadata": {},
   "source": [
    "In this example `app` has imlemented `on_new_employee`, `to_employee_email` and `to_welcome_message` functions. We can now use [Tester](../../api/fastkafka/testing/Tester/) to create their mirrored functions: `to_new_employee`, `on_employee_email` and `on_welcome_message`. "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "15592404",
   "metadata": {},
   "source": [
    "Testing process:\n",
    "\n",
    "1. `tester` produces message to the `new_employee` topic\n",
    "\n",
    "2. Assert that the `app` consumed the message from the `new_employee` topic with the accurate argument\n",
    "\n",
    "3. Within `on_new_employee` function, `to_employee_email` and `to_welcome_message` functions are called - and messages are produced to the `employee_email` and `welcome_message` topic\n",
    "\n",
    "4. Assert that the `tester` consumed the message by calling `on_employee_email`\n",
    "\n",
    "5. Assert that the `tester` consumed the message by calling `on_welcome_message`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fc0d7c25",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23-07-31 10:38:40.069 [INFO] fastkafka._testing.in_memory_broker: InMemoryBroker._patch_consumers_and_producers(): Patching consumers and producers!\n",
      "23-07-31 10:38:40.070 [INFO] fastkafka._testing.in_memory_broker: InMemoryBroker starting\n",
      "23-07-31 10:38:40.070 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': ['production_server_1:9092']}'\n",
      "23-07-31 10:38:40.071 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched start() called()\n",
      "23-07-31 10:38:40.071 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': ['production_server_1:9092']}'\n",
      "23-07-31 10:38:40.072 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched start() called()\n",
      "23-07-31 10:38:40.091 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': ['production_server_1:9092']}'\n",
      "23-07-31 10:38:40.091 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched start() called()\n",
      "23-07-31 10:38:40.092 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "23-07-31 10:38:40.092 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'auto_offset_reset': 'earliest', 'max_poll_records': 100, 'bootstrap_servers': ['production_server_1:9092']}\n",
      "23-07-31 10:38:40.093 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched start() called()\n",
      "23-07-31 10:38:40.093 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "23-07-31 10:38:40.094 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched subscribe() called\n",
      "23-07-31 10:38:40.094 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer.subscribe(), subscribing to: ['new_employee']\n",
      "23-07-31 10:38:40.095 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "23-07-31 10:38:40.096 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "23-07-31 10:38:40.097 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'auto_offset_reset': 'earliest', 'max_poll_records': 100, 'bootstrap_servers': ['production_server_1:9092']}\n",
      "23-07-31 10:38:40.098 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched start() called()\n",
      "23-07-31 10:38:40.099 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "23-07-31 10:38:40.099 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched subscribe() called\n",
      "23-07-31 10:38:40.100 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer.subscribe(), subscribing to: ['employee_email']\n",
      "23-07-31 10:38:40.100 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "23-07-31 10:38:40.101 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "23-07-31 10:38:40.101 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'auto_offset_reset': 'earliest', 'max_poll_records': 100, 'bootstrap_servers': ['production_server_1:9092']}\n",
      "23-07-31 10:38:40.102 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched start() called()\n",
      "23-07-31 10:38:40.103 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "23-07-31 10:38:40.103 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched subscribe() called\n",
      "23-07-31 10:38:40.103 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer.subscribe(), subscribing to: ['welcome_message']\n",
      "23-07-31 10:38:40.104 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "23-07-31 10:38:44.092 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched stop() called\n",
      "23-07-31 10:38:44.093 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "23-07-31 10:38:44.093 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "23-07-31 10:38:44.094 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched stop() called\n",
      "23-07-31 10:38:44.094 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "23-07-31 10:38:44.095 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "23-07-31 10:38:44.095 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched stop() called\n",
      "23-07-31 10:38:44.096 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched stop() called\n",
      "23-07-31 10:38:44.096 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "23-07-31 10:38:44.096 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "23-07-31 10:38:44.097 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched stop() called\n",
      "23-07-31 10:38:44.097 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched stop() called\n",
      "23-07-31 10:38:44.097 [INFO] fastkafka._testing.in_memory_broker: InMemoryBroker stopping\n",
      "ok\n"
     ]
    }
   ],
   "source": [
    "assert app._kafka_config[\"bootstrap_servers_id\"] == \"localhost\"\n",
    "\n",
    "async with Tester(app).using_inmemory_broker(bootstrap_servers_id=\"production\") as tester:\n",
    "    assert app._kafka_config[\"bootstrap_servers_id\"] == \"production\"\n",
    "    assert tester._kafka_config[\"bootstrap_servers_id\"] == \"production\"\n",
    "\n",
    "    # produce the message to new_employee topic\n",
    "    await tester.to_new_employee(Employee(name=\"Mickey\", surname=\"Mouse\"))\n",
    "    # previous line is equal to:\n",
    "    # await tester.mirrors[app.on_new_employee](Employee(name=\"Mickey\", surname=\"Mouse\"))\n",
    "\n",
    "    # Assert app consumed the message\n",
    "    await app.awaited_mocks.on_new_employee.assert_called_with(\n",
    "        Employee(name=\"Mickey\", surname=\"Mouse\"), timeout=5\n",
    "    )\n",
    "\n",
    "    # If the the previous assert is true (on_new_employee was called),\n",
    "    # to_employee_email and to_welcome_message were called inside on_new_employee function\n",
    "\n",
    "    # Now we can check if this two messages were consumed\n",
    "    await tester.awaited_mocks.on_employee_email.assert_called(timeout=5)\n",
    "    await tester.awaited_mocks.on_welcome_message.assert_called(timeout=5)\n",
    "\n",
    "assert app._kafka_config[\"bootstrap_servers_id\"] == \"localhost\"\n",
    "\n",
    "print(\"ok\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "547d3632",
   "metadata": {},
   "source": [
    "### Final script"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9da1c56f",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23-07-31 10:38:47.045 [INFO] fastkafka._testing.in_memory_broker: InMemoryBroker._patch_consumers_and_producers(): Patching consumers and producers!\n",
      "23-07-31 10:38:47.046 [INFO] fastkafka._testing.in_memory_broker: InMemoryBroker starting\n",
      "23-07-31 10:38:47.046 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': ['production_server_1:9092']}'\n",
      "23-07-31 10:38:47.047 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched start() called()\n",
      "23-07-31 10:38:47.048 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': ['production_server_1:9092']}'\n",
      "23-07-31 10:38:47.048 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched start() called()\n",
      "23-07-31 10:38:47.067 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': ['production_server_1:9092']}'\n",
      "23-07-31 10:38:47.067 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched start() called()\n",
      "23-07-31 10:38:47.068 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "23-07-31 10:38:47.070 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'auto_offset_reset': 'earliest', 'max_poll_records': 100, 'bootstrap_servers': ['production_server_1:9092']}\n",
      "23-07-31 10:38:47.070 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched start() called()\n",
      "23-07-31 10:38:47.071 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "23-07-31 10:38:47.071 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched subscribe() called\n",
      "23-07-31 10:38:47.072 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer.subscribe(), subscribing to: ['new_employee']\n",
      "23-07-31 10:38:47.072 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "23-07-31 10:38:47.072 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "23-07-31 10:38:47.073 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'auto_offset_reset': 'earliest', 'max_poll_records': 100, 'bootstrap_servers': ['production_server_1:9092']}\n",
      "23-07-31 10:38:47.074 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched start() called()\n",
      "23-07-31 10:38:47.074 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "23-07-31 10:38:47.074 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched subscribe() called\n",
      "23-07-31 10:38:47.075 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer.subscribe(), subscribing to: ['employee_email']\n",
      "23-07-31 10:38:47.075 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "23-07-31 10:38:47.076 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "23-07-31 10:38:47.076 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'auto_offset_reset': 'earliest', 'max_poll_records': 100, 'bootstrap_servers': ['production_server_1:9092']}\n",
      "23-07-31 10:38:47.076 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched start() called()\n",
      "23-07-31 10:38:47.077 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "23-07-31 10:38:47.077 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched subscribe() called\n",
      "23-07-31 10:38:47.078 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer.subscribe(), subscribing to: ['welcome_message']\n",
      "23-07-31 10:38:47.078 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "23-07-31 10:38:51.068 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched stop() called\n",
      "23-07-31 10:38:51.069 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "23-07-31 10:38:51.069 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "23-07-31 10:38:51.070 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched stop() called\n",
      "23-07-31 10:38:51.070 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "23-07-31 10:38:51.071 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "23-07-31 10:38:51.071 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched stop() called\n",
      "23-07-31 10:38:51.072 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched stop() called\n",
      "23-07-31 10:38:51.072 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "23-07-31 10:38:51.073 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "23-07-31 10:38:51.073 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched stop() called\n",
      "23-07-31 10:38:51.074 [INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched stop() called\n",
      "23-07-31 10:38:51.074 [INFO] fastkafka._testing.in_memory_broker: InMemoryBroker stopping\n",
      "ok\n"
     ]
    }
   ],
   "source": [
    "# | notest\n",
    "\n",
    "import asyncio\n",
    "from fastkafka._application.app import FastKafka\n",
    "from fastkafka._application.tester import Tester\n",
    "from pydantic import BaseModel, Field\n",
    "from typing import Optional\n",
    "\n",
    "\n",
    "class Employee(BaseModel):\n",
    "    name: str\n",
    "    surname: str\n",
    "    email: Optional[str] = None\n",
    "\n",
    "\n",
    "class EmaiMessage(BaseModel):\n",
    "    sender: str = \"info@gmail.com\"\n",
    "    receiver: str\n",
    "    subject: str\n",
    "    message: str\n",
    "\n",
    "\n",
    "kafka_brokers = dict(localhost=[dict(url=\"server_1\", port=9092)], production=[dict(url=\"production_server_1\", port=9092)])\n",
    "app = FastKafka(kafka_brokers=kafka_brokers)\n",
    "\n",
    "\n",
    "@app.consumes()\n",
    "async def on_new_employee(msg: Employee):\n",
    "    employee = await to_employee_email(msg)\n",
    "    await to_welcome_message(employee)\n",
    "\n",
    "\n",
    "@app.produces()\n",
    "async def to_employee_email(employee: Employee) -> Employee:\n",
    "    # generate new email\n",
    "    employee.email = employee.name + \".\" + employee.surname + \"@gmail.com\"\n",
    "    return employee\n",
    "\n",
    "\n",
    "@app.produces()\n",
    "async def to_welcome_message(employee: Employee) -> EmaiMessage:\n",
    "    message = f\"Dear {employee.name},\\nWelcome to the company\"\n",
    "    return EmaiMessage(receiver=employee.email, subject=\"Welcome\", message=message)\n",
    "\n",
    "\n",
    "async def async_tests():\n",
    "    assert app._kafka_config[\"bootstrap_servers_id\"] == \"localhost\"\n",
    "    \n",
    "    async with Tester(app).using_inmemory_broker(bootstrap_servers_id=\"production\") as tester:\n",
    "        assert app._kafka_config[\"bootstrap_servers_id\"] == \"production\"\n",
    "        assert tester._kafka_config[\"bootstrap_servers_id\"] == \"production\"\n",
    "    \n",
    "        # produce the message to new_employee topic\n",
    "        await tester.to_new_employee(Employee(name=\"Mickey\", surname=\"Mouse\"))\n",
    "        # previous line is equal to:\n",
    "        # await tester.mirrors[app.on_new_employee](Employee(name=\"Mickey\", surname=\"Mouse\"))\n",
    "\n",
    "        # Assert app consumed the message\n",
    "        await app.awaited_mocks.on_new_employee.assert_called_with(\n",
    "            Employee(name=\"Mickey\", surname=\"Mouse\"), timeout=5\n",
    "        )\n",
    "\n",
    "        # If the the previous assert is true (on_new_employee was called),\n",
    "        # to_employee_email and to_welcome_message were called inside on_new_employee function\n",
    "\n",
    "        # Now we can check if this two messages were consumed\n",
    "        await tester.awaited_mocks.on_employee_email.assert_called(timeout=5)\n",
    "        await tester.awaited_mocks.on_welcome_message.assert_called(timeout=5)\n",
    "    \n",
    "    assert app._kafka_config[\"bootstrap_servers_id\"] == \"localhost\"\n",
    "    print(\"ok\")\n",
    "\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    loop = asyncio.get_event_loop()\n",
    "    loop.run_until_complete(async_tests())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dca285c3",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "python3",
   "language": "python",
   "name": "python3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
